package com.atguigu.utils;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;

// 自定义点击流数据源
// 实现SourceFunction接口，泛型是数据源中的元素类型ClickEvent
public class ClickSource implements SourceFunction<ClickEvent> {
    private boolean running = true;
    private Random random = new Random();
    private String[] usernameArray = {"Mary", "Bob", "Alice", "Liz", "John"};
    private String[] urlArray = {"./home", "./cart", "./fav", "./product?id=1", "./product?id=2"};
    // 当flink程序启动时，触发run函数的执行
    @Override
    public void run(SourceContext<ClickEvent> ctx) throws Exception {
        // 死循环，不停的发送数据
        while (running) {
            ctx.collect(new ClickEvent(
                    usernameArray[random.nextInt(usernameArray.length)],
                    urlArray[random.nextInt(urlArray.length)],
                    // 事件发生的时间，确定性的
                    Calendar.getInstance().getTimeInMillis()
            ));
            Thread.sleep(100L);
        }
    }

    // 当取消flink任务时，触发cancel的执行
    @Override
    public void cancel() {
        running = false;
    }
}
